package com.peng.my_rocketmq.producer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.peng.my_rocketmq.bean.User;

@RestController
public class ProducerController {
	
	protected final Logger log = LoggerFactory.getLogger(this.getClass());
	
	  @Resource   
      private RocketMQTemplate rocketMQTemplate;     
		@Autowired
		private ObjectMapper objectMapper;
	  @GetMapping("/syncSend")
	public void syncSend() {
		  
		  for(int i=0;i<1000;i++) {
			   User user = new User();
		         user.setId(i);
		         user.setUsername("用户syncSend"+i);
		         String json = JSON.toJSONString(user);
			 // 以同步的方式发送消息，构造器构造对象消息给指定的topic   
//			 SendResult     sendResult = rocketMQTemplate.syncSend("peng-mq-spring-test:syncTagSend", MessageBuilder.withPayload(json).build());     
		         SendResult     sendResult =rocketMQTemplate.syncSend("peng-mq-spring-test:syncTagSend", user);
		     System.out.printf("生产发送结果：=%s %n", sendResult);         
		  }
		  
	}
	  @GetMapping("/asyncSend")
	public void asyncSend() {
	     // 异步方式发送用户定义对象类型的消息，并实现回调接口SendCallback    
		  for(int i=0;i<1000;i++) {
		  User user = new User();
	         user.setId(i);
	         user.setUsername("用户asy"+i);
	     rocketMQTemplate.asyncSend("peng-mq-spring-test:asyncTagSend",user , new SendCallback() {  
	         // 实现消息发送成功的后续处理  
	         public void onSuccess(SendResult var1) {
	         System.out.printf("成功=%s %n", var1); 
	       }       
	        // 实现消息发送失败的后续处理           
	         public void onException(Throwable var1) {
	             System.out.printf("失败=%s %n", var1); 
	       }  
	     });    
	     }   

	}
	  @GetMapping("/topicSend")
	public void topicSend() {
	     // 指定topic的同时，设置tag值，以便消费端可以根据tag值进行选择性消费      
	  for(int i=0;i<5;i++) {
			  try {
				User user = new User();
				     user.setId(i);
				     user.setUsername("测试重复tagSend0"+i);
				     Map<String, Object> map=new HashMap<String, Object>();
				     map.put(RocketMQHeaders.KEYS,"key"+i);
				     
				 	Message<byte[]> build;
				
						build = MessageBuilder.withPayload(objectMapper.writeValueAsString(user).getBytes()).setHeader(RocketMQHeaders.KEYS, "key"+i).build();
						rocketMQTemplate.syncSend("peng-mq-spring-test:syncTagSend6", build);
			} catch (JsonProcessingException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	   
//	     rocketMQTemplate.convertAndSend("peng-mq-spring-test:tagSend0",user);  
	     }
	     // tag0 will not be consumer-selected 
	  //  rocketMQTemplate.convertAndSend("peng-mq-spring-test:tagSend1", "I'm from tag1");   

	}
	  @GetMapping("/transactionSend")
	public void transactionSend(int i) {
		     User user = new User();
	         user.setId(i);
	         user.setUsername("用户transactionSend"+i);
			//Message msg = MessageBuilder.withPayload("Hello RocketMQ " + i).setHeader(RocketMQHeaders.KEYS, "KEY_" + i).build();	
			//指定在@RocketMQTransactionListener中声明的txProducerGroup, 使用这个事务发布者组来发送事务消息和回查状态
	         
	         Message msg =  MessageBuilder.withPayload(user).setHeader(RocketMQHeaders.KEYS, "KEY_" + i).build();
			SendResult sendResult = rocketMQTemplate.sendMessageInTransaction("my-spring-rocketmq-transa-producer","peng-mq-spring-test:transactionTag", msg, null);
			
			log.info("transactionSend 返回结果{} ",sendResult.getSendStatus());
	
	}
	  
	  
	  /**
	   * 支持顺序发送消息
	   */
	  @GetMapping("/sendOrder")

	    public void OrderSend(){
	        rocketMQTemplate.setMessageQueueSelector( new MessageQueueSelector() {
				
				@Override
				public MessageQueue select(List<MessageQueue> mqs, org.apache.rocketmq.common.message.Message msg, Object arg) {
					  /**
		             * mqs：要发送消息的topic下的所有消息队列集合
		             * msg：发送的消息
		             * arg：发送消息时传递的参数 通过该参数指定发送到哪个队列
		             */
		            int queueNum = Integer.valueOf(String.valueOf(arg)) % mqs.size();
		            System.out.println("队列id："+queueNum+" 消息:"+new String(msg.getBody()));
		            return mqs.get(queueNum);
				}
			});
	        
	        for(int i=1;i<=100;i++){
	        	  User user = new User();
	 	         user.setId(i);
	 	         user.setUsername("用户sendOrder"+i);
	 	        Message msg =  MessageBuilder.withPayload(user).build();
	            rocketMQTemplate.syncSendOrderly("peng-mq-spring-test:sendMessOrder",msg, String.valueOf(i));
	        }
	    }

	  
	  
	  
	  
	  
	  
	
	  
	
}
